package com.boot.notes.rocketmq.transactionMsg;

import com.alibaba.fastjson.JSON;
import com.boot.common.constant.rocket.TopicConstant;
import com.boot.common.enums.OrderEnums;
import com.boot.modules.admin.service.OrderService;
import com.boot.notes.rocketmq.transactionMsg.entity.Order;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;


import java.util.concurrent.ConcurrentHashMap;

/**
 * 实现 事务消息监听
 *
 * @className: TransactionListenerImpl
 * @auther: mkq
 * @date: 2022/01/22 10:29
 * @desctiption: TODO
 * @version: 1.0
 */
@Component
@RocketMQTransactionListener(rocketMQTemplateBeanName = TopicConstant.TRANSACTION_TOPIC)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private ConcurrentHashMap<String, RocketMQLocalTransactionState> localTrans = new ConcurrentHashMap<>();

    private Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);

    @Autowired
    private OrderService orderService;

    /**
     * 步骤二：
     * 描述：mq收到事务消息后，开始执行本地事务
     *
     * @param message
     * @param arg
     * @return
     * @Transactional：开启本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        System.out.println("half消息确认。。。");
        String orderStr = new String((byte[]) message.getPayload());
        Order order = JSON.parseObject(orderStr, Order.class);
        String transId = (String) message.getHeaders().get("rocketmq_" + RocketMQHeaders.TRANSACTION_ID);
        // 更新订单状态
        int count = orderService.updateOrderState(order.getId(), OrderEnums.PAIED.getCode());
        RocketMQLocalTransactionState result = RocketMQLocalTransactionState.COMMIT;
        try {
            if (count > 0) {
                localTrans.put(transId, result);
            }
            return result;
        } catch (Exception e) {
            logger.error("发生异常，异常信息：{}", e.getMessage());
            // 发生异常，回滚事务
            orderService.updateOrderState(order.getId(), OrderEnums.CANCEL.getCode());
            result = RocketMQLocalTransactionState.ROLLBACK;
            localTrans.put(transId, result);
            return result;
        }
    }

    /**
     * 步骤四
     * 描述：mq回调检查本地事务执行情况
     *
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        System.out.println("事务消息回查。。。");
        String transId = (String) message.getHeaders().get("rocketmq_" + RocketMQHeaders.TRANSACTION_ID);
        System.out.println("transId：" + transId);
        return localTrans.get(transId);
//        logger.info(">>>> TX message listener check local transaction, message={} <<<<",message.getPayload());
//
//        String jsonString = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
//        Order order = JSON.parseObject(jsonString, Order.class);
//        // 检查本地事务
//        RocketMQLocalTransactionState result = RocketMQLocalTransactionState.COMMIT;
//        try {
//
//            List<Order> list = orderService.list(new QueryWrapper<Order>(order));
//            if(list.size()<=0){
//                result = RocketMQLocalTransactionState.UNKNOWN;
//            }
//
//        } catch (Exception e) {
//            // 异常就回滚
//            orderService.updateOrderState(order.getId(), OrderEnums.CANCEL.getCode());
//            logger.error(">>>> exception message={} <<<<",e.getMessage());
//            result = RocketMQLocalTransactionState.ROLLBACK;
//        }
//        return result;
    }
}
